package com.nosql.db.cluster;

import com.nosql.common.protocol.DBResponse;
import com.nosql.db.core.modul.Document;
import com.nosql.db.core.storage.LSMStore;
import com.nosql.db.core.wal.WALEntry;
import com.nosql.db.network.DBCommand;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;

public class ClusterManager {
    private final String nodeId;
    private List<Node> nodes = new CopyOnWriteArrayList<>();
    private final LSMStore storage;
    private Node leader; // 静态配置的Leader

    public ClusterManager(String nodeId, String role, Properties config, LSMStore storage) throws IOException {
        this.nodeId = nodeId;
        this.storage = storage;

        if ("LEADER".equals(role)) {
            this.nodes = parseNodes(config.getProperty("cluster.nodes"));
            this.leader = new Node(nodeId,
                    config.getProperty("node.host", "localhost"),
                    Integer.parseInt(config.getProperty("server.port")),
                    Node.Role.LEADER
            );
        } else {
            this.leader = discoverLeader(parseNodes(config.getProperty("cluster.seeds")));
        }
    }

        // 新增方法
        public Node getCurrentLeader() {
            return this.leader; // 直接返回Node对象
        }

    private void initializeFollowerNodes(Properties config) throws IOException {
        // 初始化Follower配置
        List<Node> seedNodes = parseNodes(config.getProperty("cluster.seeds"));
        this.leader = discoverLeader(seedNodes);
    }


    private Node discoverLeader(List<Node> seedNodes) throws IOException {
        final int maxRetries = 3;
        final int timeoutMs = 5000;
        if (seedNodes == null || seedNodes.isEmpty()) {
            throw new IllegalArgumentException("种子节点列表为空");
        }
        for (Node node : seedNodes) {
            System.out.println("尝试连接种子节点: " + node.getId() + " " +
                    node.getHost() + ":" + node.getPort());
            for (int i = 0; i < maxRetries; i++) {
                try {
                    System.out.println("[Cluster] 创建连接...");

                    Socket socket = new Socket();
                    socket.connect(new InetSocketAddress(node.getHost(), node.getPort()), timeoutMs);
                    System.out.println("[Cluster] 连接成功!");

                    ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
                    ObjectInputStream in = new ObjectInputStream(socket.getInputStream());

                    System.out.println("[Cluster] 发送LeaderQuery...");
                    out.writeObject(new DBCommand.LeaderQuery());
                    out.flush();

                    System.out.println("[Cluster] 等待响应...");
                    Object response = in.readObject();
                    System.out.println("[Cluster] 收到响应: " + response.getClass().getName());

                    if (response instanceof DBCommand.LeaderResponse) {
                        return ((DBCommand.LeaderResponse) response).getLeader();
                    }
                } catch (IOException e) {
                    System.err.println("连接尝试 " + (i+1) + "/" + maxRetries + " 失败: " +
                            e.getClass().getSimpleName() + " - " + e.getMessage());

                    if (i == maxRetries - 1) {
                        System.err.println("节点 " + node.getId() + " 最终不可达");
                    } else {
                        try {
                            Thread.sleep(100 * (long) Math.pow(2, i));
                        } catch (InterruptedException ignored) {}
                    }
                } catch (ClassNotFoundException e) {
                    throw new IOException("响应解析失败", e);
                }
            }
        }
        throw new IOException("所有种子节点均不可达");
    }

    // 处理写请求
    public void handleWrite(DBCommand command) throws IOException {
        // 1. 处理复制命令（直接写入不转发）
        if (command.getType() == DBCommand.Type.REPLICATION) {
            DBCommand.ReplicationCommand repCmd = (DBCommand.ReplicationCommand) command;
            // 跳过来源节点避免回环
            if (!repCmd.getSourceNodeId().equals(nodeId)) {
                processLocally(repCmd.getOriginalCommand());
            }
            return;
        }
        // 处理刷新命令
        if (command.getType() == DBCommand.Type.FLUSH) {
            manualFlush();
            return;
        }

        // 2. 普通命令处理
        processLocally(command);

        if (isLeader()) {
            // Leader将命令转为复制命令广播
            replicateToNodes(command);
        } else {
            forwardToLeader(command);
        }
    }

    // 数据复制到其他节点
    private void replicateToNodes(DBCommand command) {
        nodes.parallelStream().forEach(node -> {
            if (!node.getId().equals(nodeId)) {
                new Thread(() -> {
                    // 直接调用复制方法，不处理返回值
                    sendReplication(node, new DBCommand.ReplicationCommand(nodeId, command));
                }).start();  // 在 Thread 对象上调用 start()
            }
        });
    }

    public void manualFlush() throws IOException {
        storage.flushMemTable();
    }



    // 发送复制请求（简化实现）
    private boolean sendReplication(Node node, DBCommand command) {
        try (Socket socket = new Socket(node.getHost(), node.getPort())) {
            ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
            out.writeObject(command);
            return true;
        } catch (IOException e) {
            System.err.println("复制到节点失败: " + node.getId());
            return false;
        }
    }

    // 转换命令到WAL日志条目
    private WALEntry convertToWALEntry(DBCommand command) {
        if (command instanceof DBCommand.putCommand) {
            DBCommand.putCommand putCmd = (DBCommand.putCommand) command;
            return new WALEntry(
                    WALEntry.Operation.PUT,
                    putCmd.getCollection(),
                    putCmd.getId(),
                    putCmd.getData()
            );
        } else if (command instanceof DBCommand.DeleteCommand) {
            DBCommand.DeleteCommand delCmd = (DBCommand.DeleteCommand) command;
            return new WALEntry(WALEntry.Operation.DELETE,delCmd.getCollection(),delCmd.getId());
        } else if (command instanceof DBCommand.UpdateCommand) {
            DBCommand.UpdateCommand updateCmd = (DBCommand.UpdateCommand) command;
            return new WALEntry(
                    WALEntry.Operation.UPDATE,
                    updateCmd.getCollection(),
                    updateCmd.getId(),
                    updateCmd.getUpdateData()
            );
        } else if (command instanceof DBCommand.CreateCollectionCommand) {
            DBCommand.CreateCollectionCommand cteateCmd =(DBCommand.CreateCollectionCommand) command;
            return new WALEntry(WALEntry.Operation.CREATE_COLLECTION,cteateCmd.getCollection(),cteateCmd.getCollection());
        }
        throw new IllegalArgumentException("不支持的命令类型: " + command.getType());
    }

    // 转发请求到Leader节点
    private void forwardToLeader(DBCommand command) throws IOException {
        if (leader == null) {
            throw new IOException("当前无可用Leader");
        }

        try (Socket socket = new Socket(leader.getHost(), leader.getPort())) {
            ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
            ObjectInputStream in = new ObjectInputStream(socket.getInputStream());

            out.writeObject(command);
            Object response = in.readObject();

            if (response instanceof Exception) {
                throw new IOException("Leader处理失败", (Exception) response);
            }
        } catch (ClassNotFoundException e) {
            throw new IOException("响应解析失败", e);
        }
    }

    // 新增本地处理方法（添加在ClusterManager类中）
    private void processLocally(DBCommand command) throws IOException {
        if (command instanceof DBCommand.putCommand) {
            DBCommand.putCommand putCmd = (DBCommand.putCommand) command;
            storage.put(putCmd.getCollection(), putCmd.getData(), putCmd.getId());
        } else if (command instanceof DBCommand.DeleteCommand) {
            DBCommand.DeleteCommand delCmd = (DBCommand.DeleteCommand) command;
            storage.delete(delCmd.getCollection(), delCmd.getId());
        } else if (command instanceof DBCommand.CreateCollectionCommand) {
            DBCommand.CreateCollectionCommand createCmd = (DBCommand.CreateCollectionCommand) command;
            storage.createCollection(createCmd.getCollection());
        } else if (command instanceof DBCommand.UpdateCommand) {
            DBCommand.UpdateCommand updateCmd = (DBCommand.UpdateCommand) command;
            storage.update(updateCmd.getCollection(), updateCmd.getId(), updateCmd.getUpdateData());
        }
    }

    private boolean isLeader() {
        return leader != null && leader.getId().equals(nodeId);
    }

    private Properties loadConfig(String nodeId) throws IOException {
        Properties config = new Properties();
        try (InputStream input = Files.newInputStream(Paths.get(nodeId + ".properties"))) {
            config.load(input);
        }
        return config;
    }

    public List<Node> parseNodes(String nodesConfig) {
        List<Node> nodes = new ArrayList<>();
        if (nodesConfig == null || nodesConfig.isEmpty()) {
            return nodes;
        }

        for (String nodeStr : nodesConfig.split(",")) {
            String[] parts = nodeStr.trim().split(":");
            try {
                // 处理种子节点格式 (id:host:port)
                if (parts.length == 3) {
                    nodes.add(new Node(
                            parts[0],    // id
                            parts[1],    // host
                            Integer.parseInt(parts[2]), // port
                            Node.Role.FOLLOWER // 默认角色
                    ));
                }
                // 处理完整节点格式 (id:host:port:role)
                else if (parts.length == 4) {
                    nodes.add(new Node(
                            parts[0],
                            parts[1],
                            Integer.parseInt(parts[2]),
                            Node.Role.valueOf(parts[3])
                    ));
                } else {
                    System.err.println("无效节点配置: " + nodeStr);
                }
            } catch (Exception e) {
                System.err.println("解析节点配置失败: " + nodeStr);
                e.printStackTrace();
            }
        }
        return nodes;
    }

}
